package com.bw.flinkstreaming.state2.job3;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava18.com.google.common.collect.Lists;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.UUID;

/**
 *
 * 需求：相同key的单词出现三次就计算平均值
 *   hadoop
 *   hive
 *   hive
 *   hadoop
 *   spark
 *   hadoop  -> 触发计算
 *   hive    -> 触发计算
 *   hadoop
 *   hadoop
 *   hadoop  -> 触发计算
 *
 *  自定义状态实现该需求，使用MapState状态
 * */
public class MapStateCountAvg extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Double>> {

    //定义状态
    private MapState<String,Long> mapState;

    /**
     * 执行的时候会调用一次，在这里做状态初始化
     * */
    @Override
    public void open(Configuration parameters) throws Exception {
        MapStateDescriptor descriptor = new MapStateDescriptor<String,Long>(
                "mapState",
                String.class,Long.class
        );
        mapState = getRuntimeContext().getMapState(descriptor);

    }

    /**
     * value：100，20
     * */
    @Override
    public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Double>> out) throws Exception {
        //当有数据过来后，就向map中插入一条数据
        mapState.put(UUID.randomUUID().toString(),value.f1);

        //判断当前map里面的key是否有三个
        ArrayList<Long> lists = Lists.newArrayList(mapState.values());
        if (lists.size() == 3) {
            long count =0;
            double sum = 0;
            for(long ele:lists) {
                count++;
                sum += ele;
            }
            double avg = (Double)sum / count;
            out.collect(Tuple2.of(value.f0,avg));

            //清空状态信息
            mapState.clear();
        }
    }
}
